package com.xian.java.stream;

import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.RichParallelSourceFunction;
import org.apache.flink.streaming.api.functions.source.RichSourceFunction;

import java.sql.*;
import java.util.ArrayList;
import java.util.List;

/**
 * 获取mysql数据
 */
public class GetMysqlMsg {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        DataStreamSource<String> ds = env.addSource(new MysqlSourceFunction());

        ds.print();


        env.execute();

    }



}


class MysqlSourceFunction extends RichSourceFunction<String> {

    private Connection con = null;
    private Statement st = null;
    private ResultSet rs = null;

    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);
        //打开mysql连接 返回 结果集
        Class.forName("com.mysql.jdbc.Driver");
        String db = "db1";
        con = DriverManager.getConnection("jdbc:mysql://c71/" + db, "root", "123456");
        st = con.createStatement();
        //1.获取库所有表名
        rs = st.executeQuery("select * from db1.person");
    }

    @Override
    public void run(SourceContext<String> ctx) throws Exception {
        while (rs.next()) {
            int id = rs.getInt("id");
            String name = rs.getString("name");
            ctx.collect(id + "," + name);
        }
    }

    @Override
    public void cancel() {
        System.out.println("取消。。。。。");
    }

    @Override
    public void close() throws Exception {
        super.close();
        //关闭资源
        try {
            System.out.println("关闭资源。。。。。。");
            rs.close();
            st.close();
            con.close();
        } catch (SQLException e) {
            e.printStackTrace();
        }
    }
}
